package leo.peersim.paxos;

import java.util.Vector;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import leo.peersim.common.LamportEvent;
import leo.peersim.common.LamportEventComparator;
import leo.peersim.common.LogicalClock;

import org.apache.log4j.Logger;

import peersim.core.CommonState;

/**
 * Leader agent as should be in a <i>real</i> deployment, i.e. as a thread.
 * 
 * <p>TODO However it requests synchronization with the simulator engine that it is
 * single-threaded. (Maybe it is necessary to modify peersim simulator engine).</p>
 * 
 * @author leo
 * @version June 19, 2009
 */
@Deprecated
public class LeaderTh extends PaxosAgent implements Runnable {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	private String name = "Leader";
	
	private PromisedBallots promises = new PromisedBallots();
	
	private Vector<ClientRequest> c;
	
	private Vector<ClientRequest> processedRequests;
	
	//private LeaderData ldata = null;
	
	private Lock l1 = new ReentrantLock();
	private Condition c1 = l1.newCondition();
	
	private Lock l2 = new ReentrantLock();
	private Condition c2 = l2.newCondition();
	private Condition c3 = l2.newCondition();
	
	private boolean endSimulation = false;

	private LogicalClock _attemptBallot;
	
	private ClientRequest _currentClientRequest;
	
	
	
	public LeaderTh(Paxos container) {
		super(container);
		this.c = new Vector<ClientRequest>();
		this.processedRequests = new Vector<ClientRequest>();
	}
	
	/**
	 * 
	 * @param req
	 */
	public void submit(ClientRequest req) {
		this.l1.lock();
		this.c.add(req);
		this.c1.signal();
		this.l1.unlock();
	}
	
	/**
	 * 
	 * @return next {@link ClientRequest} in the Leader's queue to be atomically
	 * broadcast
	 */
	public ClientRequest dequeue() {
		this.l1.lock();
		ClientRequest req = null;
		while(c.isEmpty()) {
			try {
				this.c1.await();
				//wait();
				if(!c.isEmpty()) {
					req = c.remove(0);
					break;
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				//this.l1.unlock();
			}
		}
		this.l1.unlock();
		return req;
	}
	
	protected String getName() {
		return this.name;
	}

	@Override
	public void run() {
		logger.trace(pp() + "hello world!");
		try {
			Thread.currentThread().sleep(1000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		while(true) {
			ClientRequest c = this.dequeue();
			logger.debug(pp() + "start processing request _" + c.getOpName() + "_...");
			this.consensus(c);
		}
	}
	
	public void manage(Decide d) {
		l2.lock();
		this._currentClientRequest.decide();
		c2.signalAll();
		logger.debug("ballot " + d.getBallotId() + " has been decided!");
		l2.unlock();
	}

	private void consensus(ClientRequest cr) {
		l2.lock();
		this.processedRequests.add(cr);
		this._currentClientRequest = cr;
		
		while(!cr.hasReachedConsensus()) {
			
			this.startNewAttempt(cr);

			this._attemptBallot = container.getNewBallot();
			
			logger.debug(pp() + "attempt _" + cr.getAttempts() + "_ with ballot " + this._attemptBallot + " to atomically broadcast _" + cr.getOpName() + "_...");
			
			Prepare prepare = new Prepare(CommonState.getNode(), container.getLogicalClock(), this._attemptBallot);
			container.broadcastToGroup(prepare);
			
			try {
				c2.await();
			} catch (InterruptedException e) {
				logger.error("Interrupted wait!!");
			}
			
		}
		l2.unlock();
	}
	
	
	
	private void phase1() {
		
	}
	
	private void phase2() {
		
	}
	
	private void startNewAttempt(ClientRequest cr) {
		cr.newAttempt();
		this.promises = new PromisedBallots();
	}
	
	public void manage(Promise p) {
		logger.trace(pp()+"received promise from " + p.getSenderNode().getID() + ", latest value: "+ p.getLatestAcceptedValue());
		
		LamportEventComparator c = new LamportEventComparator();
		if (c.compare(p.getBallotId(), this._attemptBallot)!=0) {
			// received a promis from an old ballot, dropping
			//TODO
			return;
		}
		
		if (!this.promises.containsKey(p.getSenderNode())){
			
			if (!promises.hasQuorum()) {
				// collect the new promise and re-check if the quorum is reached
				this.promises.put(p.getSenderNode(), p);
				if (promises.hasQuorum()) {
					// quorum has been reached
					long proposingValue = promises.getLargestValue();
					proposingValue+=1; // in order to give a new value to the broadcast
					logger.debug(pp() + "ballot " + this._attemptBallot + " has a quorum... proposing _" + proposingValue + "_ to all...");
					
					Accept accept = new Accept(CommonState.getNode(), container.getLogicalClock(), this._attemptBallot, "", proposingValue);
					container.broadcastToGroup(accept);
				}
			} else {
				// it has already a quorum, dropping message!
				//TODO
			}
			
		} else {
			//there is already a proposal for that node
			//TODO
		}
	}
		
}
	